package com.neusoft.databus.agent.interactive.collect;

import java.util.Arrays;
import java.util.Calendar;
import java.util.DoubleSummaryStatistics;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.springframework.web.client.RestTemplate;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.bizcore.web.utils.SpringUtil;
import com.neusoft.databus.agent.pool.ptrest.PtrestConnector;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.bean.MetricBean;
import com.neusoft.databus.common.bean.ResourceBean;
import com.neusoft.databus.common.connection.PorttrafficConnection;
import com.neusoft.databus.common.exception.ConfigErrorException;
import com.neusoft.databus.common.interactive.collect.Collector;
import com.neusoft.databus.common.parser.Parser;

import lombok.extern.slf4j.Slf4j;

/**
 * description
 *
 * @author sunchf
 * @date 2019年4月2日 上午9:37:26
 */
@Slf4j
public class PorttrafficPtrestCollector extends Collector<RestTemplate> {

    private static final String FLOWS2 = "flows";
    private static final String NAME2 = "name";
    private static final String BANDWIDTH_UR = "bandwidthUR";
    private static final String PORT_BANDWIDTH = "portBandwidth";
    private static final String OUT_FLOW_AVG_UR = "outFlowAvgUR";
    private static final String OUT_FLOW_HIGH_UR = "outFlowHighUR";
    private static final String OUT_FLOW_AVG = "outFlowAvg";
    private static final String OUT_FLOW_LOW = "outFlowLow";
    private static final String OUT_FLOW_HIGH = "outFlowHigh";
    private static final String IN_FLOW_AVG_UR = "inFlowAvgUR";
    private static final String IN_FLOW_HIGH_UR = "inFlowHighUR";
    private static final String IN_FLOW_AVG = "inFlowAvg";
    private static final String IN_FLOW_LOW = "inFlowLow";
    private static final String IN_FLOW_HIGH = "inFlowHigh";
    private static final String RSP_CODE = "rspCode";

    private static List<String> sumValueMatricList = null;
    private static List<String> avgValueMatricList = null;
    static {
        PorttrafficPtrestCollector.sumValueMatricList =
                Arrays.asList(PorttrafficPtrestCollector.IN_FLOW_LOW, PorttrafficPtrestCollector.IN_FLOW_HIGH,
                        PorttrafficPtrestCollector.IN_FLOW_AVG,
                        PorttrafficPtrestCollector.OUT_FLOW_LOW,
                        PorttrafficPtrestCollector.OUT_FLOW_HIGH, PorttrafficPtrestCollector.OUT_FLOW_AVG,
                        PorttrafficPtrestCollector.PORT_BANDWIDTH);

        PorttrafficPtrestCollector.avgValueMatricList =
                Arrays.asList(PorttrafficPtrestCollector.IN_FLOW_HIGH_UR,
                        PorttrafficPtrestCollector.IN_FLOW_AVG_UR,
                        PorttrafficPtrestCollector.OUT_FLOW_HIGH_UR, PorttrafficPtrestCollector.OUT_FLOW_AVG_UR,
                        PorttrafficPtrestCollector.BANDWIDTH_UR);
    }

    public PorttrafficPtrestCollector(Parser parser) {
        super(parser);
        this.connector = SpringUtil.getBean(PtrestConnector.class);
    }

    @Override
    protected void doCollect(CommandBean command) throws Exception {

        if (command.getResourceBean().size() != 1) {
            throw new ConfigErrorException("不支持批量采集");
        }

        final List<MetricBean> metrics = command.getResourceBean().get(0).getMetrics();
        if (metrics.size() == 0) {
            throw new ConfigErrorException("端口流量采集需要配置指标");
        }

        try {
            this.conn = this.connector.getConn(command.getConnection().asUsedConnection());
            this.sendCommand(command);

        } catch (final Exception e) {
            throw e;
        } finally {
            try {
                if (PorttrafficPtrestCollector.log.isDebugEnabled()) {
                    PorttrafficPtrestCollector.log.debug("return connection object to poll, {}",
                            command.getConnection());
                }
                this.connector.returnConn(command.getConnection().asUsedConnection(), this.conn);
            } catch (final Exception e) {

            }
        }

    }

    /**
     * 参照文档《IDC获取端口流量数据20220705-修订版.docx》
     *
     * @param commandBean
     */
    @SuppressWarnings("unchecked")
    public void sendCommand(CommandBean commandBean) {

        final ResourceBean resource = commandBean.getResourceBean().get(0);
        if (PorttrafficPtrestCollector.log.isInfoEnabled()) {
            PorttrafficPtrestCollector.log.info("begin to communicate with port traffic server");
        }
        final PorttrafficConnection connection = (PorttrafficConnection) commandBean.getConnection();

        final String url = connection.getUrl();
        final Map<String, Object> config = resource.getConfig();
        // interval作为参数融合进资源的扩展配置中。在查询API中，按照interval配置时间，向前查找几分钟

        final int interval = config.containsKey("interval") ? Integer.valueOf((String) config.get("interval")) : 5;

        final Calendar cal = Calendar.getInstance();
        final String endTime = DateFormatUtils.format(cal, "yyyy-MM-dd HH:mm:ss");
        cal.add(Calendar.MINUTE, -interval);
        final String startTime = DateFormatUtils.format(cal, "yyyy-MM-dd HH:mm:ss");

        // 该参数是在DataBus资源扩展配置中配置
        final String portIds = (String) config.get("portIds");
        final String[] portIdList = portIds.split(",");

        final Map<String, List<Object>> mutiplePortTrafficValue = Maps.newHashMap();

        for (final String portId : portIdList) {
            // 拼接端口流量的远程地址
            final String requestUrl =
                    url + "?portTrafficId=" + portId + "&startTime=" + startTime + "&endTime=" + endTime;
            if (PorttrafficPtrestCollector.log.isDebugEnabled()) {
                PorttrafficPtrestCollector.log.debug("IDC获取端口流量数据，请求URL：{}", requestUrl);
            }

            final String resultStr = this.conn.getForObject(requestUrl, String.class);
            if (PorttrafficPtrestCollector.log.isDebugEnabled()) {
                PorttrafficPtrestCollector.log.debug("IDC获取端口流量数据，返回结果：{}", resultStr);
            }
            final Map<String, Object> resultMap = JsonUtils.jsonToMap(resultStr);
            if (resultMap.containsKey(PorttrafficPtrestCollector.RSP_CODE)
                    && "0".equals(resultMap.get(PorttrafficPtrestCollector.RSP_CODE))) {
                final Map<String, Object> portTrafficDataMap = (Map<String, Object>) resultMap.get("portTrafficData");
                final double inFlowHigh =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.IN_FLOW_HIGH));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.IN_FLOW_HIGH)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.IN_FLOW_HIGH).add(inFlowHigh);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.IN_FLOW_HIGH,
                            Lists.newArrayList(inFlowHigh));
                }
                final double inFlowLow =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.IN_FLOW_LOW));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.IN_FLOW_LOW)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.IN_FLOW_LOW).add(inFlowLow);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.IN_FLOW_LOW, Lists.newArrayList(inFlowLow));
                }
                final double inFlowAvg =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.IN_FLOW_AVG));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.IN_FLOW_AVG)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.IN_FLOW_AVG).add(inFlowAvg);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.IN_FLOW_AVG, Lists.newArrayList(inFlowAvg));
                }
                final double inFlowHighUR =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.IN_FLOW_HIGH_UR));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.IN_FLOW_HIGH_UR)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.IN_FLOW_HIGH_UR).add(inFlowHighUR);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.IN_FLOW_HIGH_UR,
                            Lists.newArrayList(inFlowHighUR));
                }
                final double inFlowAvgUR =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.IN_FLOW_AVG_UR));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.IN_FLOW_AVG_UR)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.IN_FLOW_AVG_UR).add(inFlowAvgUR);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.IN_FLOW_AVG_UR,
                            Lists.newArrayList(inFlowAvgUR));
                }
                final double outFlowHigh =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.OUT_FLOW_HIGH));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.OUT_FLOW_HIGH)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.OUT_FLOW_HIGH).add(outFlowHigh);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.OUT_FLOW_HIGH,
                            Lists.newArrayList(outFlowHigh));
                }
                final double outFlowLow =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.OUT_FLOW_LOW));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.OUT_FLOW_LOW)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.OUT_FLOW_LOW).add(outFlowLow);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.OUT_FLOW_LOW,
                            Lists.newArrayList(outFlowLow));
                }
                final double outFlowAvg =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.OUT_FLOW_AVG));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.OUT_FLOW_AVG)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.OUT_FLOW_AVG).add(outFlowAvg);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.OUT_FLOW_AVG,
                            Lists.newArrayList(outFlowAvg));
                }
                final double outFlowHighUR =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.OUT_FLOW_HIGH_UR));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.OUT_FLOW_HIGH_UR)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.OUT_FLOW_HIGH_UR).add(outFlowHighUR);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.OUT_FLOW_HIGH_UR,
                            Lists.newArrayList(outFlowHighUR));
                }
                final double outFlowAvgUR =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.OUT_FLOW_AVG_UR));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.OUT_FLOW_AVG_UR)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.OUT_FLOW_AVG_UR).add(outFlowAvgUR);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.OUT_FLOW_AVG_UR,
                            Lists.newArrayList(outFlowAvgUR));
                }
                final double portBandwidth =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.PORT_BANDWIDTH));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.PORT_BANDWIDTH)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.PORT_BANDWIDTH).add(portBandwidth);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.PORT_BANDWIDTH,
                            Lists.newArrayList(portBandwidth));
                }
                final double bandwidthUR =
                        this.getDoubleValue(portTrafficDataMap.get(PorttrafficPtrestCollector.BANDWIDTH_UR));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.BANDWIDTH_UR)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.BANDWIDTH_UR).add(bandwidthUR);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.BANDWIDTH_UR,
                            Lists.newArrayList(bandwidthUR));
                }
                final String name = (String) portTrafficDataMap.get(PorttrafficPtrestCollector.NAME2);
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.NAME2)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.NAME2).add(name);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.NAME2, Lists.newArrayList(name));
                }
                final String flows = JsonUtils
                        .pojoToJson(
                                (List<Map<String, Object>>) portTrafficDataMap.get(PorttrafficPtrestCollector.FLOWS2));
                if (mutiplePortTrafficValue.containsKey(PorttrafficPtrestCollector.FLOWS2)) {
                    mutiplePortTrafficValue.get(PorttrafficPtrestCollector.FLOWS2).add(flows);
                } else {
                    mutiplePortTrafficValue.put(PorttrafficPtrestCollector.FLOWS2, Lists.newArrayList(flows));
                }

            }
        }

        if (mutiplePortTrafficValue.isEmpty()) {
            final String msg = "数据拉取失败，请检查系统运行日志";
            PorttrafficPtrestCollector.log.warn(msg);
            commandBean.setErrorCode(msg);
            return;
        }

        // 说明拉取的多个端口的数据，求和平均处理
        if (mutiplePortTrafficValue.get(PorttrafficPtrestCollector.IN_FLOW_HIGH).size() > 0) {
            for (final Map.Entry<String, List<Object>> entry : mutiplePortTrafficValue.entrySet()) {
                if (PorttrafficPtrestCollector.sumValueMatricList.contains(entry.getKey())) {
                    //final Object minValue =
                    //        Collections.max(entry.getValue(), (a, b) -> (Double) a <= (Double) b ? -1 : 1);
                    //mutiplePortTrafficValue.put(entry.getKey(), Arrays.asList(minValue));

                    final Object summmaryStatistics = entry.getValue().stream().map(it -> this.getDoubleValue(it))
                            .collect(Collectors.summarizingDouble(Double::doubleValue));

                    mutiplePortTrafficValue.put(entry.getKey(),
                            Arrays.asList(((DoubleSummaryStatistics) summmaryStatistics).getSum()));
                } else if (PorttrafficPtrestCollector.avgValueMatricList.contains(entry.getKey())) {

                    final Object avgValue = entry.getValue().stream().map(it -> this.getDoubleValue(it))
                            .collect(Collectors.averagingDouble(Double::doubleValue));
                    mutiplePortTrafficValue.put(entry.getKey(), Arrays.asList(avgValue));
                }
            }
        }

        if (PorttrafficPtrestCollector.log.isDebugEnabled()) {
            PorttrafficPtrestCollector.log.debug("parsed mutiple port traffic value: {}",
                    JsonUtils.pojoToJson(mutiplePortTrafficValue));
        }

        resource.getMetrics().forEach(m -> {
            m.setOriginalValue(mutiplePortTrafficValue.containsKey(m.getVariable())
                    ? mutiplePortTrafficValue.get(m.getVariable()).get(0) : null);
            m.setValue(mutiplePortTrafficValue.containsKey(m.getVariable())
                    ? mutiplePortTrafficValue.get(m.getVariable()).get(0) : null);
        });

    }

    private double getDoubleValue(Object val) {
        return Double.valueOf(val.toString());
    }
}
